package rdd01.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark_Jihe {
  def main(args: Array[String]): Unit = {
    val sparkConf=new SparkConf().setMaster("local[*]").setAppName("FlatMap")
    val sc=new SparkContext(sparkConf)
    val rdd1=sc.makeRDD(List(1,2,3,4))
    val rdd2=sc.makeRDD(List(3,4,5,6))
    //交集
    val rddjaio: RDD[Int] = rdd1.intersection(rdd2)
    println(rddjaio.collect().mkString(","))
    //并集
    val rddUnion:RDD[Int]=rdd1.union(rdd2)
//    rddUnion.saveAsTextFile("union")
    println(rddUnion.collect().mkString(" union "))
    //差集（交叉重复的去掉，留下不重复的）
    val rddSubtract:RDD[Int]=rdd1.subtract(rdd2)
    println(rddSubtract.collect().mkString(" ,"))
    //拉链:将两个数据源中的数据以key-value形似进行合并，第一个集合中的值作为key,第二个集合中的值作为value
    val rddZip: RDD[(Int, Int)] = rdd1.zip(rdd2)
    println(rddZip.collect().mkString(" zip "))
    val rdd5=sc.makeRDD(List(("a",1),("a",2),("b",3),("b",2)))
    //reduceByKey():相同的key数据进行value数据聚合操作
    val reduceRDD: RDD[(String, Int)] = rdd5.reduceByKey((x: Int, y: Int) => {
     println(s"x=${x},y=${y}")
      x + y
    })
    reduceRDD.collect()
      .foreach(println)
    println("=================")
    //groupByKey():将数据源中的数据，相同key的数据分在一个组中，形成一个对偶元组
    //元组中的第一个元素就是key，元组中的第二个元素就是相同key的value的集合
    val groupRDD: RDD[(String, Iterable[Int])] = rdd5.groupByKey()
    groupRDD.collect().foreach(println)
     sc.stop()
  }

}
